package com.centnet.base.bigdata.hbase;


import com.google.common.collect.Lists;
import com.inspinia.base.util.RandomValue;
import org.apache.commons.collections4.map.HashedMap;

import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.RecursiveTask;

/**
 * Created by veblen on 16/6/7.
 */
public class BatchHbaseDataTask extends RecursiveTask<Long> {
    String tableName = "source_data_phone";
    String cf = "if";
    Random random = new Random();
    private long total;

    private int THRESHOLD = 10000;

    public BatchHbaseDataTask(long total) {
        this.total = total;
    }

    @Override
    protected Long compute() {

        long sum = 0;
        if (total <= THRESHOLD) {
            Map<String, List<HBaseColumn>> rowColumns = new HashedMap();
            for (int i = 0; i < total; i++) {
                String coller = "13008840769";
                HBaseColumn caller = new HBaseColumn(cf, "caller", coller, String.class);
                HBaseColumn called = new HBaseColumn(cf, "called", RandomValue.getPhone(), String.class);
                HBaseColumn callDuration = new HBaseColumn(cf, "callDuration", random.nextInt(200) + "", Integer.class);

                Date date = RandomValue.getDate("2016-03-01", "2016-06-01");
                String row = coller + "00" + date.getTime();
                List<HBaseColumn> list = Lists.newArrayList(caller, callDuration, called);
                rowColumns.put(row, list);
            }
            long star = System.currentTimeMillis();
            HBaseUtil.batchPut(tableName, rowColumns);
            long end = System.currentTimeMillis();
            System.out.println(end - star + "ms");
            return Long.valueOf(rowColumns.size());
        } else {
            float t = total / THRESHOLD;
            long partition = (long) Math.floor(t);

            for (int i = 0; i < partition; i++) {
                long size;
                if (i != (partition - 1)) {
                    size = total / partition;
                } else {
                    size = total - ((total / partition) * i);
                }
                if (size > 0) {
                    BatchHbaseDataTask task = new BatchHbaseDataTask(size);
                    task.fork();
                    sum += task.join();
                }
            }
            return sum;
        }
    }
}
